package cn.com.charles.flink.drools.flatmap;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;
import org.drools.core.definitions.InternalKnowledgePackage;
import org.drools.core.impl.KnowledgeBaseImpl;
import org.kie.api.KieBaseConfiguration;
import org.kie.api.KieServices;
import org.kie.api.conf.EventProcessingOption;
import org.kie.api.definition.rule.Rule;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.EntryPoint;
import org.kie.internal.builder.KnowledgeBuilder;
import org.kie.internal.builder.KnowledgeBuilderFactory;
import org.kie.internal.io.ResourceFactory;
import org.kie.internal.utils.KieHelper;

import cn.com.charles.flink.drools.domain.AlarmObj;
import cn.com.charles.flink.drools.domain.RuleObj;

public class RuleAlarmFunction extends RichCoFlatMapFunction<AlarmObj, RuleObj, AlarmObj> {
	private static final long serialVersionUID = -1454325317652872205L;
	private KieHelper kieHelper;
	private KnowledgeBaseImpl kieBase;
	private KieSession kieSession;
	private EntryPoint entryPoint;
	private Lock lock = new ReentrantLock();

	@Override
	public void open(Configuration parameters) throws Exception {
		if (kieSession == null) {
			kieHelper = new KieHelper();
			KieBaseConfiguration config = KieServices.Factory.get().newKieBaseConfiguration();
			config.setOption(EventProcessingOption.STREAM);
			try {
				kieBase = (KnowledgeBaseImpl) kieHelper.build();
				kieSession = kieBase.newStatefulSession();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public void close() throws Exception {
		Thread.sleep(1000);
		kieSession.destroy();
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public void flatMap1(AlarmObj alarm, Collector collector) throws Exception {
		lock.lock();
		try {
			entryPoint = kieSession.getEntryPoint("Demo02");
			if (entryPoint != null) {
				entryPoint.insert(alarm);
				int fireNum = kieSession.fireAllRules();
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
		collector.collect(alarm);
	}

	@SuppressWarnings({ "rawtypes" })
	@Override
	public void flatMap2(RuleObj rule, Collector collector) throws Exception {
		if (rule.getOperate() == 0) {
			lock.lock();
			try {
				kieBase.removeRule("rules", rule.getName());
				kieSession = kieBase.newKieSession();
			} finally {
				lock.unlock();
			}
		} else {
			// 规则名是ruleId
			Rule ruleCache = kieBase.getRule("rules", rule.getName());

			// 规则如果已存在，不需要再次新增
			if (ruleCache != null && rule.getName().equals(ruleCache.getName())) {
				System.out.println("规则已经存在，ruleName=" + ruleCache.getName());
			} else {
				//重新添加规则
				KnowledgeBuilder kb = KnowledgeBuilderFactory.newKnowledgeBuilder();
				//装入规则，可以装入多个
				kb.add(ResourceFactory.newByteArrayResource(rule.getDrlStr().getBytes("utf-8")), ResourceType.DRL);

				kb.getKnowledgePackages().forEach(k -> {
					kieBase.addPackage((InternalKnowledgePackage) k);
				});

				Rule rule1 = kieSession.getKieBase().getRule("rules", "rule2");
				System.out.println("rule3:=" + rule1);

				//				kieBase.addKnowledgePackages(kb.getKnowledgePackages());
				kieSession = kieBase.newKieSession();
			}
		}
	}

}